<?php

namespace app\common\controller;

use think\Controller;
use think\Env;

class Kafka extends Controller
{
    /**
     * kafka生产者
     *
     * @return \think\Response
     */
    public function logProducer($topicName,$data)
    {
        $partition = 0;
        $rk = new \RdKafka\Producer();
        $kafakaGroup = \Env::get('kafka.group');
        if($kafakaGroup == 'cqyq_qn'){
            $topicName .= '_qn';
        }
        $rk->addBrokers(\Env::get('kafka.host'));
        $topic = $rk->newTopic($topicName);
        $topic->produce(RD_KAFKA_PARTITION_UA,$partition,json_encode($data));
        if ($rk->getOutQLen() > 0) {
            $rk->poll(500);
        } else {
            $rk->poll(0);
        }
    }

    public function logConsumer($topic){
        $conf = new \RdKafka\Conf();
        $conf->set('group.id', \Env::get('kafka.group'));
        $conf->set('bootstrap.servers', \Env::get('kafka.host'));
        $rk = new \RdKafka\KafkaConsumer($conf);
        self::$consume = $rk;
        self::$consume->subscribe($topic);
        while (true) {
            $message = self::$consume->consume(1000);
            switch ($message->err) {
                case RD_KAFKA_RESP_ERR_NO_ERROR:
                    echo '接收到消息: ' . $message->payload . PHP_EOL;
                    break;
                case RD_KAFKA_RESP_ERR__PARTITION_EOF:
                    echo '已达到分区末尾，没有更多消息可消费' . PHP_EOL;
                    break;
                case RD_KAFKA_RESP_ERR__TIMED_OUT:
                     echo '消费超时' . PHP_EOL;
                    break;
                default:
                    echo '消费错误: ' . $message->errstr() . PHP_EOL;
                    break;
            }
        }
    }


}
